package com.mjf.day7

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Kafka, Schema}

/**
 * Table执行环境：
 *  老版本planner批流执行环境是分开的
 *  blink planner批流执行环境是一体的
 */
object TableApiExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    // 默认创建的老版本的env，相当与1.1
    //    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 1.创建表环境
    // 1.1创建老版本的流查询环境
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    // 1.2创建老版本的批式查询环境
    val batchEnv: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    val batchTableEnv: BatchTableEnvironment = BatchTableEnvironment.create(batchEnv)

    // 1.3创建blink版本的流查询环境
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, bsSettings)

    // 1.4创建blink版本的批查询环境
    val bbSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inBatchMode()
      .build()
//    val bbTableEnv: TableEnvironment = TableEnvironment.create(bbSettings)


    // 2.从外部系统读取数据，在环境中注册表
    val filePath = "D:\\coding\\idea\\flink-stu\\src\\main\\input\\sensor.txt"

    tableEnv
      // 2.1连接到文件系统（csv）
      .connect(new FileSystem().path(filePath))
      // 定义读取数据之后的格式化方法
//      .withFormat(new OldCsv()) // 新版本会失效
      // 使用新标准版Csv，需要导入对应依赖
      .withFormat(new Csv())
      // 定义表结构
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      // 注册一张表
      .createTemporaryTable("inputTable")

    val inputTable: Table = tableEnv.sqlQuery("select * from inputTable")
    inputTable.toAppendStream[(String, Long, Double)].print("file")

    // 2.2连接到Kafka
    tableEnv
      .connect(
        new Kafka()
          .version("0.11")
          .topic("sensor")
          .property("bootstrap.servers", "hadoop103:9092")
          .property("zookeeper.connect", "hadoop103:2181")
      )
      .withFormat(new Csv())  // Kafka必须是新版的
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE())
      )
      .createTemporaryTable("kafkaInputTable")

    val kafkaSensorTable: Table = tableEnv.sqlQuery("select * from kafkaInputTable")
    kafkaSensorTable.toAppendStream[(String, Long, Double)].print("kafka")

    // 将结果写出到表
//    kafkaSensorTable.insertInto("")


    // 3.表的查询
    // 3.1简单查询，过滤投影
    val sensorTable: Table = tableEnv.from("inputTable")
    val resultTable: Table = sensorTable
      .select("id, temperature")
      .filter("id = 'sensor_1'")

    // 3.2SQL简单查询
    val resultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select id,temperature
        |from inputTable
        |where id = 'sensor_1'
        |""".stripMargin)

    // 3.3简单聚合，统计每个传感器温度个数
    val aggResultTable: Table = sensorTable
      .groupBy("id")
      .select('id, 'id.count as 'count) // 表达式写法

    // 3.4SQL简单聚合
    val aggResultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select id,count(1) as num
        |from inputTable
        |group by id
        |""".stripMargin)

    // 聚合结果需要用`toRetractStream`输出
    aggResultSqlTable.toRetractStream[(String, Long)].print("agg")

    env.execute("TableApiExample")

  }
}
